package com.teamone.etl.taskmanager.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.teamone.etl.exception.CommonCode;
import com.teamone.etl.exception.CustomException;
import com.teamone.etl.taskmanager.bo.TaskProcess;
import com.teamone.etl.taskmanager.dao.*;
import com.teamone.etl.taskmanager.entity.*;
import com.teamone.etl.taskmanager.feign.DataSourceFeignService;
import com.teamone.etl.taskmanager.service.ParseProcessJsonService;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Resource;

import org.springframework.context.annotation.Scope;
import org.springframework.context.annotation.ScopedProxyMode;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.context.WebApplicationContext;

/**
 * @Author: SixJin
 * @Date: 2021/1/12 8:40
 */
@Service
@Scope(value = WebApplicationContext.SCOPE_REQUEST, proxyMode = ScopedProxyMode.INTERFACES)
@Transactional(rollbackFor = Exception.class)
public class ParseProcessJsonServiceImpl implements ParseProcessJsonService {
    @Resource
    private EtlTaskInfoDao taskInfoDao;
    @Resource
    private EtlTaskLineInfoDao taskLineInfoDao;
    @Resource
    private EtlTaskInputInfoDao taskInputInfoDao;
    @Resource
    private EtlTaskJoinInfoDao taskJoinInfoDao;
    @Resource
    private EtlTaskGroupInfoDao taskGroupInfoDao;
    @Resource
    private EtlTaskOrderInfoDao taskOrderInfoDao;
    @Resource
    private EtlTaskLogicInfoDao taskLogicInfoDao;
    @Resource
    private EtlTaskOutputInfoDao taskOutputInfoDao;
    @Resource
    private DataSourceFeignService dataSourceFeignService;


    private final TaskProcess TASK_PROCESS = new TaskProcess();

    @Override
    public void parseJson(String taskProcessJson) {
        JSONObject taskProcessJsonObject = JSONObject.parseObject(taskProcessJson);

        JSONObject taskProcessBean = taskProcessJsonObject.getJSONObject("taskProcessBean");
        JSONObject processJson = taskProcessJsonObject.getJSONObject("processJson");

        TASK_PROCESS.setTaskId(taskProcessBean.getString("taskId"));

        this.parseLineList(processJson.getJSONArray("lineList"));
        this.parseNodeList(processJson, true);

        EtlTaskInfo etlTaskInfo = taskInfoDao.selectById(TASK_PROCESS.getTaskId());
        etlTaskInfo.setProcessJson(taskProcessJson);
        taskInfoDao.updateById(etlTaskInfo);
        TASK_PROCESS.setEtlTaskInfo(etlTaskInfo);
    }

    public void parseNodeList(JSONObject processJson, boolean flag) {
        List<JSONObject> nodeList = processJson.getJSONArray("nodeList").toJavaList(JSONObject.class);
        Date now = new Date();
        AtomicInteger index = new AtomicInteger(0);
        nodeList.forEach(node -> {
            String type = node.getString("type");
            switch (type) {
                case "table":
                    TASK_PROCESS.getTaskInputInfos().add(new EtlTaskInputInfo(node, now));
                    TASK_PROCESS.setDatasourceId(node.getJSONObject("data").getJSONArray("selectId").getJSONObject(0).getString("id"));
                    break;
                case "join":
                    TASK_PROCESS.getTaskJoinInfos().add(new EtlTaskJoinInfo(node, now,
                            TASK_PROCESS.getTaskInputInfos().get(index.get()).getId(),
                            TASK_PROCESS.getTaskInputInfos().get(index.get() + 1).getId()));
                    break;
                case "group":
                    TASK_PROCESS.getTaskGroupInfos().add(new EtlTaskGroupInfo(node, now));
                    break;
                case "order":
                    TASK_PROCESS.getTaskOrderInfos().add(new EtlTaskOrderInfo(node, now));
                    break;
                case "logic":
                    TASK_PROCESS.getTaskLogicInfos().add(new EtlTaskLogicInfo(node, now));
                    break;
                case "output":
                    TASK_PROCESS.getTaskOutputInfos().add(new EtlTaskOutputInfo(node, now));
                    TASK_PROCESS.setOutputDatasourceId(node.getJSONObject("data").getJSONArray("selectId").getJSONObject(0).getString("id"));
                    TASK_PROCESS.setOutputDatabaseName(node.getJSONObject("data").getString("databaseName"));
                    break;
                case "start":
                case "end":
                default:
                    break;
            }
        });
        if (flag) {
            saveNodeInfo();
        }
    }

    public void saveNodeInfo() {
        HashMap<String, Object> columnMap = new HashMap<>(1);
        columnMap.put("task_id", TASK_PROCESS.getTaskId());
        taskLineInfoDao.deleteByMap(columnMap);
        taskInputInfoDao.deleteByMap(columnMap);
        taskJoinInfoDao.deleteByMap(columnMap);
        taskGroupInfoDao.deleteByMap(columnMap);
        taskOrderInfoDao.deleteByMap(columnMap);
        taskLogicInfoDao.deleteByMap(columnMap);
        taskOutputInfoDao.deleteByMap(columnMap);
        TASK_PROCESS.getTaskLineInfos().forEach(taskLine -> taskLineInfoDao.insert(taskLine));
        TASK_PROCESS.getTaskInputInfos().forEach(taskInput -> taskInputInfoDao.insert(taskInput));
        TASK_PROCESS.getTaskJoinInfos().forEach(taskJoin -> taskJoinInfoDao.insert(taskJoin));
        TASK_PROCESS.getTaskGroupInfos().forEach(taskGroup -> taskGroupInfoDao.insert(taskGroup));
        TASK_PROCESS.getTaskOrderInfos().forEach(taskOrder -> taskOrderInfoDao.insert(taskOrder));
        TASK_PROCESS.getTaskLogicInfos().forEach(taskLogic -> taskLogicInfoDao.insert(taskLogic));
        TASK_PROCESS.getTaskOutputInfos().forEach(taskOutput -> taskOutputInfoDao.insert(taskOutput));
    }


    public void parseLineList(JSONArray lineList) {
        for (int i = 0; i < lineList.size(); i++) {
            EtlTaskLineInfo taskLine = lineList.getJSONObject(i).toJavaObject(EtlTaskLineInfo.class);
            taskLine.setId(UUID.randomUUID().toString());
            taskLine.setTaskId(TASK_PROCESS.getTaskId());
            TASK_PROCESS.getTaskLineInfos().add(taskLine);
        }
    }

    private final StringBuffer TASK_SQL = new StringBuffer("select ");

    @Override
    public String getTask(String taskId) throws CustomException {
        EtlTaskInfo task = taskInfoDao.selectById(taskId);
        String taskProcessJson = task.getProcessJson();
        if (taskProcessJson == null) {
            throw new CustomException(CommonCode.TASK_PROCESS_NULL);
        }
        parseNodeList(JSONObject.parseObject(taskProcessJson).getJSONObject("processJson"), false);
        TASK_PROCESS.getTaskInputInfos().forEach(taskInput -> {
            String[] tableFields = taskInput.getTableField().split(",");
            String[] tableAsField = taskInput.getTableAsField().split(",");
            String tableAsName = taskInput.getTableAsName();
            String tableName = tableAsName != null ? tableAsName : taskInput.getDatabaseName() + "." + taskInput.getTableName();
            for (int i = 0; i < tableFields.length; i++) {
                TASK_SQL.append(tableName).append(".").append(tableFields[i]).append(" ").append(tableAsField[i]).append(",");
            }
        });
        TASK_SQL.deleteCharAt(TASK_SQL.length() - 1);
        appendJoin();
        appendGroup();
        appendOrder();
        executeSql();
        return TASK_SQL.toString();
    }

    public void appendJoin() {
        List<EtlTaskJoinInfo> taskJoinInfos = TASK_PROCESS.getTaskJoinInfos();
        List<EtlTaskInputInfo> taskInputInfos = TASK_PROCESS.getTaskInputInfos();
        if (taskJoinInfos.size() > 0) {
            for (int i = 0; i < taskJoinInfos.size(); i++) {
                EtlTaskInputInfo left = taskInputInfos.get(i);
                EtlTaskInputInfo right = taskInputInfos.get(i + 1);
                TASK_SQL.append(" from ").append(left.getDatabaseName()).append(".").append(left.getTableName());
                if (left.getTableAsName() != null) {
                    TASK_SQL.append(" as ").append(left.getTableAsName());
                }
                switch (taskJoinInfos.get(i).getJoinType()) {
                    case "1":
                    case "left":TASK_SQL.append(" left join ");break;
                    case "2":
                    case "right":TASK_SQL.append(" right join ");break;
                    case "3":
                    case "inner":TASK_SQL.append(" inner join ");break;
                    default:
                        break;
                }
                TASK_SQL.append(right.getDatabaseName()).append(".").append(right.getTableName());
                if (right.getTableAsName() != null) {
                    TASK_SQL.append(" as ").append(right.getTableAsName());
                }
                TASK_SQL.append(" on ");
                String leftTableName = left.getTableAsName() == null ? left.getTableName() : left.getTableAsName();
                String rightTableName = right.getTableName() == null ? right.getTableName() : right.getTableAsName();
                TASK_SQL.append(leftTableName).append(".").append(taskJoinInfos.get(i).getLeftJoinField())
                        .append("=").append(rightTableName).append(".").append(taskJoinInfos.get(i).getRightJoinField());
            }
        } else {
            EtlTaskInputInfo input = taskInputInfos.get(0);
            TASK_SQL.append(" from ").append(input.getDatabaseName()).append(".").append(input.getTableName());
        }
    }

    public void appendGroup() {
        List<EtlTaskGroupInfo> taskGroupInfos = TASK_PROCESS.getTaskGroupInfos();
        List<EtlTaskInputInfo> taskInputInfos = TASK_PROCESS.getTaskInputInfos();
        if (taskGroupInfos.size() > 0) {
            for (int i = 0; i < taskGroupInfos.size(); i++) {
                if (i == 0) {
                    TASK_SQL.append(" group by ");
                }
                int finalI = i;
                taskInputInfos.forEach(taskInput -> {
                    if (taskInput.getNodeId().equals(taskGroupInfos.get(finalI).getSelectNodeId())) {
                        String tableName = taskInput.getTableAsName() == null ? taskInput.getDatabaseName() + "." + taskInput.getTableName() : taskInput.getTableAsName();
                        TASK_SQL.append(tableName).append(".").append(taskGroupInfos.get(finalI).getSelectTableField()).append(",");
                    }
                });
            }
            TASK_SQL.deleteCharAt(TASK_SQL.length() - 1);
        }

    }

    public void appendOrder() {
        List<EtlTaskOrderInfo> taskOrderInfos = TASK_PROCESS.getTaskOrderInfos();
        List<EtlTaskInputInfo> taskInputInfos = TASK_PROCESS.getTaskInputInfos();
        if (taskOrderInfos.size() > 0) {
            for (int i = 0; i < taskOrderInfos.size(); i++) {
                if (i == 0) {
                    TASK_SQL.append(" order by ");
                }
                int finalI = i;
                taskInputInfos.forEach(taskInput -> {
                    if (taskInput.getNodeId().equals(taskOrderInfos.get(finalI).getSelectNodeId())) {
                        String tableName = taskInput.getTableAsName() == null ? taskInput.getDatabaseName() + "." + taskInput.getTableName() : taskInput.getTableAsName();
                        TASK_SQL.append(tableName).append(".").append(taskOrderInfos.get(finalI).getSelectTableField());
                    }
                });
                if (taskOrderInfos.get(i).getOrderSort().equals(1)) {
                    TASK_SQL.append(" asc,");
                } else {
                    TASK_SQL.append(" desc,");
                }
            }
            TASK_SQL.deleteCharAt(TASK_SQL.length() - 1);
        }
    }

    public void executeSql() throws CustomException {
        if (TASK_PROCESS.getTaskOutputInfos() == null) {
            return;
        }
        EtlTaskOutputInfo taskOutputInfo = TASK_PROCESS.getTaskOutputInfos().get(0);
        String[] inputFields = taskOutputInfo.getInputField().split(",");
        String[] outputFields = taskOutputInfo.getOutputTableField().split(",");
        if (inputFields.length == 0 || outputFields.length == 0) {
            throw new CustomException(CommonCode.INPUT_OR_OUTPUT_FIELD_NULL);
        }
        boolean shouldExecuteRuleEngine = TASK_PROCESS.getTaskLogicInfos().size() > 0;
        Map<String, Object> data = dataSourceFeignService.dataAnalysis(
                Integer.parseInt(TASK_PROCESS.getDatasourceId()),
                TASK_PROCESS.getTaskInputInfos().get(0).getDatabaseName(),
                TASK_PROCESS.getTaskInputInfos().get(0).getTableName(),
                TASK_SQL.toString(),
                shouldExecuteRuleEngine).getData();
        StringBuffer insertSql = new StringBuffer("insert into ").append(taskOutputInfo.getOutputTable()).append("(");
        for (String outputField : outputFields) {
            insertSql.append(outputField).append(",");
        }
        insertSql.deleteCharAt(insertSql.length() - 1).append(")").append(" values");
        List<Map<String, String>> maps = JSON.parseObject(JSON.toJSONString(data.get("results")),
                new TypeReference<List<Map<String, String>>>() {});
        maps.forEach(map -> {
            insertSql.append("(");
            for (String column : map.keySet()) {
                boolean hasColumn = false;
                for (String inputField : inputFields) {
                    if (column.equals(inputField)) {
                        hasColumn = true;
                        break;
                    }
                }
                if (hasColumn) {
                    insertSql.append("'").append(map.get(column)).append("'").append(",");
                }
            }
            insertSql.deleteCharAt(insertSql.length() - 1);
            insertSql.append("),");
        });
        insertSql.deleteCharAt(insertSql.length() - 1);
        dataSourceFeignService.dataAnalysis(
                Integer.parseInt(TASK_PROCESS.getOutputDatasourceId()),
                TASK_PROCESS.getOutputDatabaseName(),
                TASK_PROCESS.getTaskOutputInfos().get(0).getOutputTable(),
                insertSql.toString(), false);
    }

}
